package com.fanxl.flink.datastream.windows;


import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author Dell
 * @Title:
 * @Description: 请写注释类
 * @date 2022/6/29
 */
public class CustomEventTimeTrigger extends Trigger<Object, TimeWindow> {


    /**
     * 最后一个元素到达时间
     */


    @Override
    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        long start = window.getStart();
        long end = window.getEnd();
        long maxTimestamp = window.maxTimestamp();

        long currentWatermark = ctx.getCurrentWatermark();

        if(timestamp >= end){
            //到达元素的时间，大于窗口的结束时间则触发
//            System.err.println("窗口范围timestamp > end="+start+"e="+end+"时间="+timestamp);
            return TriggerResult.FIRE;
        }else if(start <= timestamp && timestamp< end){
            //还在窗口中就不触发
//            System.err.println("窗口范围start <= timestamp && timestamp< end > end   s= "+start+"  e="+end+" 时间= "+timestamp +" 窗口最大时间戳"+maxTimestamp);
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }else if (maxTimestamp <= currentWatermark) {
            // if the watermark is already past the window fire immediately
//            System.err.println("窗口范围maxTimestamp <= currentWatermark ="+start+"e="+end+"时间="+timestamp);
            return TriggerResult.FIRE;
        } else {
//            System.err.println("窗口范围s=  "+start+"e=  "+end+"时间="+timestamp +"窗口最大时间戳"+maxTimestamp);
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }

    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.deleteEventTimeTimer(window.maxTimestamp());
    }

    public static CustomEventTimeTrigger create() {
        return new CustomEventTimeTrigger();
    }

}
